﻿// Copyright (c) 2012 - 2021 Jared Zheng <jaredz at outlook dot com>
// Buddha Framework is licensed under Mulan PSL v2.
// You can use this software according to the terms and conditions of the Mulan PSL v2.
// You may obtain a copy of Mulan PSL v2 at:
//          http://license.coscl.org.cn/MulanPSL2
// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
// EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
// MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
// See the Mulan PSL v2 for more details.
//
// Create : 2012-12-01 version 0.1 Jared Zheng <jaredz at outlook dot com>
// Update : 2021-03-19 version 4.0.0 Jared Zheng <jaredz at outlook dot com>

#include "stdafx.h"
#include "SlaveServer.h"

///////////////////////////////////////////////////////////////////////////////////////////////////////////
// CSlaveServer
#define PAK_HEAD( ename, uevent ) CPAKHead ename((UEVENT)uevent, (URANK)m_SrvData.uRank, (USID)m_SrvData.usId)
#define PAK_DATA( ename, uevent, udata ) CPAKHead ename((UEVENT)uevent, (URANK)m_SrvData.uRank, (USID)m_SrvData.usId, (UDATA)udata)

#define NET_HEAD( ename, uevent ) CNETHead ename((UEVENT)uevent, (URANK)m_SrvData.uRank, (USID)m_SrvData.usId)
#define NET_DATA( ename, uevent, udata ) CNETHead ename((UEVENT)uevent, (URANK)m_SrvData.uRank, (USID)m_SrvData.usId, (UDATA)udata)

#define SYNCACK_HEAD( ename ) CNETSync ename((UEVENT)PAKE_SYNCACK, (URANK)m_SrvData.uRank, (USID)m_SrvData.usId)
#define SYNCACK_DATA( ename, udata ) CNETSync ename((UEVENT)PAKE_SYNCACK, (URANK)m_SrvData.uRank, (USID)m_SrvData.usId, (UDATA)udata)

UInt CSlaveServer::Command(PCXStr, uintptr_t)
{
    return (UInt)RET_OKAY;
}

UInt CSlaveServer::Update(void)
{
    if (m_SrvData.nStatus < SERVERS_ERROR)
    {
        // network fail : error
        if (m_NetworkPtr->Check(m_pConfig->GetUpdateTick() * COMMONC_CHECK))
        {
            m_SrvData.SetMayDay(); // 网络工作线程异常
            m_SrvData.nStatus = SERVERS_ERROR;
            LOG_ERROR(m_FileLog, TF("[Slave]网络线程异常, 进入异常状态"));
            return RET_FAIL;
        }
        if (Events.GetEventQueuePtr()->CheckThreads(m_SrvData.uQCheck))
        {
            m_SrvData.SetMayDay(); // 事件队列线程异常
            m_SrvData.nStatus = SERVERS_ERROR;
            LOG_ERROR(m_FileLog, TF("[Slave]事件队列线程异常, 进入异常状态"));
            return RET_FAIL;
        }
        m_RoutinePtr->Update();
        // tcp listen : warn
        CheckListen();
        CheckUpdate();
        SyncUpdate();
        return RET_OKAY;
    }
    return RET_FAIL;
}

void CSlaveServer::CheckListen(void)
{
    if (m_SrvData.sListen == 0)
    {
        if (StartListen() && m_SrvData.IsWarnLoad())
        {
            LOG_WARN(m_FileLog, TF("[Slave]监听重开, 尝试消除告警状态"));
            m_FileLog.Flush();
            m_SrvData.Clear();
            if (m_SrvData.IsWarnLoad(false) == false)
            {
                m_SrvData.usLoad += m_pConfig->GetServerLoad(m_SrvData.uRank);
            }
        }
    }
}

void CSlaveServer::CheckUpdate(void)
{
    bool bLive = false;
    if (m_SrvData.usUpdate == (UShort)SERVERU_SYNC)
    {
        m_nLiveUpdate = 0;
    }
    else if (m_nLiveStep > 0)
    {
        ++m_nLiveUpdate;
        if (m_nLiveUpdate >= m_nLiveStep)
        {
            m_nLiveUpdate = 0;
            bLive = true;
            //LOG_INFO(m_FileLog, TF("[Slave]服务没有状态更新, 发送心跳数据包"));
        }
    }

    for (UInt i = COMMONR_MASTER; i < COMMONR_MAX; ++i)
    {
        if (m_SrvData.nLink[i] != SERVERL_NONE)
        {
            CheckConnect(i);
            CheckUpdate(i, bLive);
        }
    }

    if (m_SrvData.usUpdate == (UShort)SERVERU_SYNC)
    {
        LOG_INFO(m_FileLog, TF("[Slave]状态更新, 通知管理更新数据"));
        m_SrvData.usUpdate = (UShort)SERVERU_OKAY;
        m_MgrHandler->OnHandle(PAKE_UPDATEACK, reinterpret_cast<uintptr_t>(&m_SrvData), m_SrvData.uRank);
    }
}

bool CSlaveServer::CheckConnect(UInt uRank)
{
    Int& nLink = m_SrvData.nLink[uRank];
    if (nLink < SERVERL_CNNTED)
    {
        if (nLink == SERVERL_READY)
        {
            StartConnect(uRank);
        }
        else
        {
            --nLink;
        }
    }
    else if (nLink < SERVERL_LINKED)
    {
        if (nLink == SERVERL_CNNTED)
        {
            LinkServer(uRank);
        }
        else
        {
            --nLink;
        }
    }
    return true;
}

void CSlaveServer::CheckUpdate(UInt uRank, bool bLive)
{
    if (m_SrvData.usUpdate == (UShort)SERVERU_SYNC)
    {
        if (m_SrvData.nLink[uRank] == SERVERL_LINKED)
        {
            NET_DATA(Update, PAKE_UPDATE, COMMONT_SLAVE);
            if (Update.Alloc(m_NetworkPtr))
            {
                CStream& Stream = Update.GetStream(false);
                m_SrvData.Status(Stream);

                LOGV_INFO(m_FileLog, TF("[Slave]向连接的内部服务(RANK=%d)更新服务状态"), uRank);
                CNetStreamPtr StreamPtr;
                Update.GetBuffer(m_NetworkPtr, StreamPtr);
                m_NetworkPtr->Send(m_SrvData.sConnect[uRank], StreamPtr);
            }
            else
            {
                LOGV_WARN(m_FileLog, TF("[Slave]向连接的内部服务(RANK=%d)更新服务状态申请缓存失败"), uRank);
                m_FileLog.Flush();
            }
        }
    }
    else if (bLive)
    {
        if (m_SrvData.sConnect[uRank] != 0)
        {
            PAK_DATA(Live, PAKE_LIVE, COMMONT_SLAVE);
            m_NetworkPtr->Send(m_SrvData.sConnect[uRank], Live);
        }
    }
}

void CSlaveServer::SyncUpdate(void)
{
    for (UInt i = COMMONR_MIN; i < COMMONR_MAX; ++i)
    {
        SyncServer(i);
    }
}

void CSlaveServer::SyncServer(UInt uRank)
{
    SERVER_CLUSTER& Cluster = SyncCluster(uRank);
    if (Cluster.Status.usUpdate == (UShort)SERVERU_SYNC)
    {
        ARY_SYNC aSocket;
        bool bRet = m_Sync[uRank].GetSync(aSocket);
        {
            CSyncLockWaitScope scope(Cluster.Lock);
            if (bRet)
            {
                PINDEX index = Cluster.Info.GetFirstIndex();
                while (index != nullptr)
                {
                    bRet = SyncServerPack(Cluster, index, aSocket, uRank);
                }
                LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)的集群状态向订阅的服务(数量=%d)同步服务信息完成"), uRank, aSocket.GetSize());
            }
            else
            {
                bRet = true;
                LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)的集群状态更新, 没有订阅的服务, 直接更新为完成状态"), uRank);
            }
            if (bRet)
            {
                Cluster.Update();
                Cluster.Status.usUpdate = (UShort)SERVERU_OKAY;
            }
        }
        if (bRet)
        {
            LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)的集群状态更新, 通知管理更新数据"), uRank);
            m_MgrHandler->OnHandle(PAKE_SYNC, reinterpret_cast<uintptr_t>(&Cluster.Status), uRank);
        }
    }
    SERVER_CLUSTER& scLocal = m_Cluster[uRank];
    if (scLocal.Status.usUpdate == (UShort)SERVERU_SYNC)
    {
        CSyncLockWaitScope scope(scLocal.Lock);
        scLocal.Update();
        scLocal.Status.usUpdate = (UShort)SERVERU_OKAY;
    }
    SERVER_CLUSTER& scMaster = m_Master[uRank];
    if (scMaster.Status.usUpdate == (UShort)SERVERU_SYNC)
    {
        CSyncLockWaitScope scope(scMaster.Lock);
        scMaster.Update();
        scMaster.Status.usUpdate = (UShort)SERVERU_OKAY;
    }
}

bool CSlaveServer::SyncServerPack(SERVER_CLUSTER& Cluster, PINDEX& index, ARY_SYNC& aSocket, UInt uRank)
{
    SYNCACK_DATA(SyncAck, uRank);
    if (SyncAck.Alloc(m_NetworkPtr, true))
    {
        CStream& Stream = SyncAck.GetStream(false);

        UInt uCount = 0;
        while (Cluster.CheckWrite(uCount, index, Stream));

        if (uCount > 0)
        {
            SyncAck.SetStatus(Cluster.Status);
            SERVER_STATUS& Status = SyncAck.GetStatus();
            Status.usRate   = (UShort)uCount; // 当前同步的数量
            Status.usUpdate = (index != nullptr) ? (UShort)SERVERU_NONE : (UShort)SERVERU_SYNC;

            LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)的集群状态向订阅的服务(数量=%d)同步%d服务信息"), uRank, aSocket.GetSize(), uCount);
            CNetStreamPtr StreamPtr;
            SyncAck.GetBuffer(m_NetworkPtr, StreamPtr);
            m_NetworkPtr->Send(aSocket, StreamPtr);
        }
        else
        {
            LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)的集群状态向订阅的服务(数量=%d)同步服务信息为空"), uRank, aSocket.GetSize());
        }
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Slave](RANK=%d)的集群状态向订阅的服务(数量=%d)同步服务信息申请缓存失败"), uRank, aSocket.GetSize());
        m_FileLog.Flush();
    }
    return true;
}
//--------------------------------------
UInt CSlaveServer::Init(CEventHandler& EventHandlerRef, CServerConfig& ConfigRef, UInt uRank)
{
    if (m_SrvData.nStatus == SERVERS_NONE)
    {
        // 创建服务日志
        if (m_FileLog.IsValid() == false)
        {
            m_FileLog.SetLevel(ConfigRef.GetFrameLogLevel());
            m_FileLog.Create(ConfigRef.GetLogPath(), ConfigRef.GetServerName(uRank));
        }
        LOG_INFO(m_FileLog, TF("[Slave]========================================"));
        LOG_INFO(m_FileLog, TF("[Slave]初始化处理开始!"));
        m_MgrHandler    = &EventHandlerRef;
        m_pConfig       = &ConfigRef;
        m_SrvData.uRank = uRank;
        assert((m_SrvData.uRank > COMMONR_MASTER) && (m_SrvData.uRank < COMMONR_MAX));
        // 从配置加载数据
        if (LoadConfig() == false)
        {
            return (UInt)RET_FAIL;
        }
        // 按配置信息加载业务层
        if (LoadRoutine() == false)
        {
            return (UInt)RET_FAIL;
        }
        m_SrvData.nStatus = SERVERS_INIT;
        LOG_INFO(m_FileLog, TF("[Slave]初始化处理完成!"));
        return (UInt)RET_OKAY;
    }
    return (UInt)RET_FAIL;
}

bool CSlaveServer::LoadConfig(void)
{
    // 配置信息输出到日志
    assert(m_pConfig != nullptr);
    m_pConfig->Dump(m_FileLog, m_SrvData.uRank);
    // 从动态变量表获取网络层智能指针
    assert(m_NetworkPtr == nullptr);
    m_NetworkPtr = m_pConfig->GetKeyValue().GetPtrValue<CNetwork>(CServerConfig::NetworkPtr);
    if (m_NetworkPtr == nullptr)
    {
        LOG_ERROR(m_FileLog, TF("[Slave]从动态变量表获取网络层智能指针无效"));
        m_FileLog.Flush();
        return false;
    }
    m_nLiveStep   = m_pConfig->GetTimeoutStep();
    m_nLiveUpdate = 0;
    LOGV_INFO(m_FileLog, TF("[Slave]加载参数LiveStep=%d"), m_nLiveStep);

    // 从配置加载数据
    assert(m_SrvData.sListen == 0);
    const SERVER_CONFIG& sc = m_pConfig->GetServerConfig(m_SrvData.uRank);
    m_SrvData.usId   = sc.usId;
    //m_SrvData.usLoad = sc.usLoad;
    m_SrvData.usRate = sc.usRate;
    m_SrvData.SetOverLoad(); // 注册完成提供正常服务

    // 标记链接对应层级服务
    for (UInt i = COMMONR_MASTER; i < COMMONR_MAX; ++i)
    {
        if ((sc.Connect[i].usAttr != 0) && (i != m_SrvData.uRank))
        {
            m_SrvData.nLink[i] = SERVERL_READY;
        }
    }

    m_SrvData.uQCheck = (UInt)(m_pConfig->GetUpdateTick() * COMMONC_CHECK);
    return true;
}

bool CSlaveServer::LoadRoutine(void)
{
    assert(m_RoutinePtr == nullptr);

    const SERVER_CONFIG& sc = m_pConfig->GetServerConfig(m_SrvData.uRank);
    if (CObjectTraits::Load<ICommonRoutine>(sc.strInst, sc.strModule.GetBuffer(), m_RoutinePtr, sc.strRoutine.GetBuffer()) == false)
    {
        LOGV_ERROR(m_FileLog, TF("[Slave]加载业务层Routine=%s, Inst=%s, Module=%s失败"),
                   sc.strRoutine.GetBuffer(), sc.strInst.GetBuffer(), sc.strModule.GetBuffer());
        m_FileLog.Flush();
        return false;
    }
    LOGV_INFO(m_FileLog, TF("[Slave]加载业务层Routine=%s, Inst=%s, Module=%s完成"),
              sc.strRoutine.GetBuffer(), sc.strInst.GetBuffer(), sc.strModule.GetBuffer());
    if (m_RoutinePtr->Load(this, m_SrvData.uRank))
    {
        // 加载服务信息m_SrvData扩展数据
        // uSync=(1 << COMMON_RANK_ENUM)的组合, 表示同步其他服务的标志位
        // StrAddr, strRoutine额外的地址信息和业务附加数据
        Events.SetThreads((Int)m_pConfig->GetServerQueue(m_SrvData.uRank));
        Events.SetThreadEvent(PAKE_QUEUESTART, PAKE_QUEUESTOP);
        return true;
    }
    return false;
}
//--------------------------------------
void CSlaveServer::Exit(void)
{
    if (m_SrvData.nStatus != SERVERS_NONE)
    {
        m_SrvData.nStatus = SERVERS_NONE;

        LOG_INFO(m_FileLog, TF("[Slave]结束处理开始!"));
        UnloadRoutine();
        UnloadConfig();

        m_pConfig    = nullptr;
        m_MgrHandler = nullptr;
        LOG_INFO(m_FileLog, TF("[Slave]结束处理完成!"));
    }
}

void CSlaveServer::UnloadConfig(void)
{
    m_SrvData.Reset();
    m_SrvData.ResetData();

    m_NetworkPtr = nullptr;
}

void CSlaveServer::UnloadRoutine(void)
{
    Events.Exit();
    if (m_RoutinePtr != nullptr)
    {
        m_RoutinePtr->Unload();
        m_RoutinePtr = nullptr;
    }
}
//--------------------------------------
bool CSlaveServer::Start(void)
{
    if (m_SrvData.nStatus == SERVERS_INIT)
    {
        LOG_INFO(m_FileLog, TF("[Slave]开启处理开始!"));
        Events.Init();
        if (CObjectTraits::Connect(Events, *m_RoutinePtr) == nullptr)
        {
            return false;
        }
        if (Events.StartQueue() == false)
        {
            return false;
        }
        if (m_RoutinePtr->Start() == false)
        {
            return false;
        }
        if (StartListen() == false)
        {
            return false;
        }
        if (StartConnect() == false)
        {
            return false;
        }

        m_SrvData.nStatus = SERVERS_RUN;
        LOG_INFO(m_FileLog, TF("[Slave]开启处理完成!"));
    }
    return true;
}

bool CSlaveServer::StartListen(void)
{
    const SERVER_CONFIG& sc = m_pConfig->GetServerConfig(m_SrvData.uRank);
    if (m_SrvData.sListen != 0)
    {
        LOGV_INFO(m_FileLog, TF("[Slave]监听%s:%d地址的Socket已经存在"), sc.Listen.strIp.GetBuffer(), sc.Listen.usPort);
        return true;
    }
    if (m_pConfig->GetAnyListen() != 0)
    {
        m_SrvData.sListen = m_NetworkPtr->Create(*this, sc.Listen.usPort);
    }
    else
    {
        m_SrvData.sListen = m_NetworkPtr->Create(*this, sc.Listen.usPort, sc.Listen.strIp.GetBuffer());
    }
    bool bRet = false;
    if (m_SrvData.sListen != 0)
    {
        bRet = m_NetworkPtr->Listen(m_SrvData.sListen);
        LOGV_INFO(m_FileLog, TF("[Slave]监听%s:%d%s"), sc.Listen.strIp.GetBuffer(), sc.Listen.usPort, bRet ? TF("成功") : TF("失败"));
        if (bRet)
        {
            Events(PAKE_TCPLISTEN, (uintptr_t)m_SrvData.sListen, m_SrvData.uRank);
            m_MgrHandler->OnHandle(PAKE_TCPLISTEN, (uintptr_t)m_SrvData.sListen, m_SrvData.uRank);
        }
    }
    else
    {
        LOGV_ERROR(m_FileLog, TF("[Slave]创建%s:%d的Socket失败"), sc.Listen.strIp.GetBuffer(), sc.Listen.usPort);
        m_FileLog.Flush();
    }
    return bRet;
}

bool CSlaveServer::StartConnect(void)
{
    for (UInt i = COMMONR_MASTER; i < COMMONR_MAX; ++i)
    {
        if (m_SrvData.nLink[i] == SERVERL_READY)
        {
            if (StartConnect(i) == false)
            {
                return false;
            }
        }
    }
    return true;
}

bool CSlaveServer::StartConnect(UInt uRank)
{
    const SERVER_CONFIG& scConnect = m_pConfig->GetServerConfig(m_SrvData.uRank);

    Int& nLink = m_SrvData.nLink[uRank];
    Socket_t& sConnect = m_SrvData.sConnect[uRank];

    if (sConnect == 0)
    {
        sConnect = m_NetworkPtr->Create(*this, scConnect.Connect[uRank].usPort, scConnect.Connect[uRank].strIp.GetBuffer());
    }
    bool bRet = (sConnect != 0);
    if (bRet)
    {
        m_NetworkPtr->SetAttr(sConnect, uRank);

        LOGV_INFO(m_FileLog, TF("[Slave]创建连接内部服务(RANK=%d)的Socket成功"), uRank);
        const SERVER_CONFIG& scListen = m_pConfig->GetServerConfig(uRank);

        assert(nLink == SERVERL_READY);
        nLink = SERVERL_CNNTING;
        bRet = m_NetworkPtr->Connect(sConnect, scListen.Listen.usPort, scListen.Listen.strIp.GetBuffer());
        LOGV_INFO(m_FileLog, TF("[Slave]连接内部服务(RANK=%d)[%s]:%d请求%s"), uRank, scListen.Listen.strIp.GetBuffer(), scListen.Listen.usPort, bRet ? TF("成功") : TF("失败"));
    }
    else
    {
        LOGV_ERROR(m_FileLog, TF("[Slave]创建连接内部服务(RANK=%d)的Socket失败"), uRank);
        m_FileLog.Flush();
    }
    return bRet;
}
//--------------------------------------
void CSlaveServer::Stop(bool bUnlink)
{
    if (m_SrvData.nStatus > SERVERS_INIT)
    {
        if (bUnlink)
        {
            CAtomics::Exchange<UInt>((PUInt)&m_SrvData.nStatus, SERVERS_UNLINK);
            for (UInt i = COMMONR_MASTER; i < COMMONR_MAX; ++i)
            {
                UnlinkServer(i);
            }
            m_RoutinePtr->Stop();
            Events.Exit();
        }
        else
        {
            LOG_INFO(m_FileLog, TF("[Slave]停止处理开始!"));
            CAtomics::Exchange<UInt>((PUInt)&m_SrvData.nStatus, SERVERS_STOP);
            StopListen();
            StopConnect();

            for (UInt i = COMMONR_MIN; i < COMMONR_MAX; ++i)
            {
                m_Master[i].Info.RemoveAll();
                m_Master[i].Reset();

                m_Cluster[i].Info.RemoveAll();
                m_Cluster[i].Reset();

                m_Sync[i].Reset();
            }

            m_SrvData.Zero();
            m_SrvData.usLoad = m_pConfig->GetServerLoad(m_SrvData.uRank);

            m_SrvData.nStatus = SERVERS_INIT;
            LOG_INFO(m_FileLog, TF("[Slave]停止处理完成!"));
        }
    }
}

void CSlaveServer::StopListen(void)
{
    if (m_SrvData.sListen != 0)
    {
        m_NetworkPtr->Destroy(m_SrvData.sListen);
        m_SrvData.sListen = 0;
        LOG_INFO(m_FileLog, TF("[Slave]停止监听Socket"));
    }
}

void CSlaveServer::StopConnect(void)
{
    for (UInt i = COMMONR_MASTER; i < COMMONR_MAX; ++i)
    {
        if (m_SrvData.nLink[i] != SERVERL_NONE)
        {
            StopConnect(i);
        }
    }
}

void CSlaveServer::StopConnect(UInt uRank)
{
    Socket_t& sConnect = m_SrvData.sConnect[uRank];
    if (sConnect != 0)
    {
        m_NetworkPtr->Destroy(sConnect);
        sConnect = 0;
        LOGV_INFO(m_FileLog, TF("[Slave]停止连接内部服务(RANK=%d)的Socket"), uRank);
    }
    m_SrvData.nLink[uRank] = SERVERL_NONE;
}
//--------------------------------------
bool CSlaveServer::OnTcpDispatch(const PacketPtr& PktPtr, PTCP_PARAM pTcp)
{
    // 1. PTCP_PARAM->pCache != nullptr时上层可以保存网络层申请的缓存数据
    //    PTCP_PARAM->pCache, PTCP_PARAM->index是网络层申请的缓存数据的起始地址和索引号
    // 2. PTCP_PARAM->pData为当前数据开始地址, PTCP_PARAM->stSize为当前数据大小
    // 3. 如果上层保存缓存, 设置PTCP_PARAM->pData为nullptr通知网络层不要再使用当前缓存
    // 4. 释放缓存使用MObject::sMCFree(index, pCache)
    bool bRet  = true;
    UInt uRank = PktPtr->GetRank();
    if ((uRank >= COMMONR_MASTER) && (uRank < COMMONR_MAX))
    {
        switch (PktPtr->GetEvent())
        {
        case PAKE_LIVE:
            {
                // 统计在线服务数量, 心跳时将在线数量回复给对应的服务
                SERVER_STATUS& ss = m_Cluster[uRank].Status;
                PAK_DATA(LiveAck, PAKE_LIVEACK, ss.uOnline);
                m_NetworkPtr->Send(pTcp->sSocket, LiveAck);
            }
            break;
        case PAKE_LIVEACK:
        case PAKE_UPDATEACK:
            {
                // 需要处理的通用信令, 不需要传递给业务层
                // 心跳/更新返回数据是在线数量, 保存主同步服务的数值
                if (uRank == COMMONR_MASTER)
                {
                    m_Cluster[m_SrvData.uRank].Status.uOnline = PktPtr->GetData();
                }
            }
            break;
        case PAKE_LINK:
        case PAKE_LINKACK:
        case PAKE_UPDATE:
        case PAKE_SYNC:
        case PAKE_SYNCACK:
        case PAKE_UNLINK:
        case PAKE_UNLINKACK:
            {
                // 需要处理的通用信令, 并且需要传递给业务层
                bRet = OnCommonEvent(uRank, PktPtr, pTcp->sSocket);
                if (bRet == false)
                {
                    break;
                }
            }
            //break;
        default:
            {
                if (PktPtr->IsStream())
                {
                    bRet = PktPtr->Bind(m_NetworkPtr, pTcp);
                }
                if (bRet)
                {
                    Events(PktPtr->GetEvent(), *PktPtr, (ULLong)pTcp->sSocket);
                }
            }
        }
    }
    else
    {
        bRet = false;
        LOGV_WARN(m_FileLog, TF("[Slave]无法识别的层级编号=%d, 信令=%d, 信令包数据大小=%d, 链接=%p"),
                  uRank, PktPtr->GetEvent(), pTcp->stSize, pTcp->sSocket);
        m_FileLog.Flush();
    }
    return bRet;
}

bool CSlaveServer::OnCommonEvent(UInt uRank, const PacketPtr& PktPtr, Socket_t sSocket)
{
    bool bRet = true;
    switch (PktPtr->GetEvent())
    {
    case PAKE_LINK:
        {
            CNETHead* pLink = static_cast<CNETHead*>(PktPtr.Get());
            bRet = OnLink(uRank, sSocket, pLink);
        }
        break;
    case PAKE_LINKACK:
        {
            CPAKHead* pLinkAck = static_cast<CPAKHead*>(PktPtr.Get());
            OnLinkAck(uRank, pLinkAck, sSocket);
        }
        break;
    case PAKE_UPDATE:
        {
            CNETHead* pUpdate = static_cast<CNETHead*>(PktPtr.Get());
            bRet = OnUpdate(uRank, sSocket, pUpdate);
        }
        break;
    case PAKE_SYNC:
        {
            CPAKHead* pSync = static_cast<CPAKHead*>(PktPtr.Get());
            bRet = OnSync(uRank, sSocket, pSync);
        }
        break;
    case PAKE_SYNCACK:
        {
            // 保存主同步服务的同步信息
            if (uRank == COMMONR_MASTER)
            {
                CNETSync* pSyncAck = static_cast<CNETSync*>(PktPtr.Get());
                OnSyncAck(uRank, pSyncAck, sSocket);
            }
        }
        break;
    case PAKE_UNLINK:
        {
            CPAKHead* pHead = static_cast<CPAKHead*>(PktPtr.Get());
            bRet = OnUnlink(uRank, sSocket, pHead);
        }
        break;
    case PAKE_UNLINKACK:
        {
            OnUnlinkAck(uRank, sSocket);
        }
        break;
    default:{}
    }
    return bRet;
}

bool CSlaveServer::OnLink(UInt uRank, Socket_t sSocket, CNETHead* pLink)
{
    bool bRet = LinkServer(uRank, sSocket, pLink);
    if (bRet)
    {
        // type[high-32, slave-rank-16 -- link-type-16]
        ULLong ullLiveData = ((ULLong)m_SrvData.uRank) << 48;
        ullLiveData += ((ULLong)pLink->GetData()) << 32;
        ullLiveData += uRank; // rank[low-32]

        m_NetworkPtr->SetAttr(sSocket, ullLiveData);

        // 统计在线服务数量, 将在线数量回复给对应的服务
        SERVER_STATUS& ss = m_Cluster[uRank].Status;
        PAK_DATA(LinkAck, PAKE_LINKACK, ss.uOnline);
        bRet = m_NetworkPtr->Send(sSocket, LinkAck);
    }
    if (bRet == false)
    {
        m_NetworkPtr->Destroy(sSocket);
    }
    return bRet;
}

bool CSlaveServer::OnUpdate(UInt uRank, Socket_t sSocket, CNETHead* pUpdate)
{
    bool bRet = UpdateServer(uRank, sSocket, pUpdate);
    if (bRet)
    {
        // 统计在线服务数量, 将在线数量回复给对应的服务
        SERVER_STATUS& ss = m_Cluster[uRank].Status;
        PAK_DATA(UpdateAck, PAKE_UPDATEACK, ss.uOnline);
        bRet = m_NetworkPtr->Send(sSocket, UpdateAck);
    }
    if (bRet == false)
    {
        m_NetworkPtr->Destroy(sSocket);
    }
    return bRet;
}

bool CSlaveServer::OnSync(UInt uRank, Socket_t sSocket, CPAKHead* pSync)
{
    UInt uComb = pSync->GetData(); // 包含需要同步的加载类型组合
    if (uComb != 0)
    {
        LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)要求同步的层级比特掩码%x"), uRank, uComb);
        for (UInt i = COMMONR_MIN; i < uRank; ++i)
        {
            UInt uSync = (1 << i);
            if ((uComb & uSync) != 0)
            {
                SyncServer(i, sSocket, pSync);
            }
        }
    }
    return true;
}

bool CSlaveServer::OnUnlink(UInt uRank, Socket_t sSocket, CPAKHead* pHead)
{
    bool bRet = UnlinkServer(uRank, sSocket, pHead->GetData());
    if (bRet)
    {
        m_NetworkPtr->SetAttr(sSocket, 0);
        PAK_HEAD(UnlinkAck, PAKE_UNLINKACK);
        bRet = m_NetworkPtr->Send(sSocket, UnlinkAck);
    }
    if (bRet == false)
    {
        m_NetworkPtr->Destroy(sSocket);
    }
    return bRet;
}

bool CSlaveServer::LinkServer(UInt uRank, Socket_t sSocket, CNETHead* pLink)
{
    assert(pLink->IsTempStream());
    CStream& Stream = pLink->GetTempStream();

    PAIR_INFO info(sSocket);
    info.m_V.Status(Stream);
    assert(info.m_V.usUpdate == (UShort)SERVERU_LINK);
    info.m_V.Info(Stream);

    UInt uRet = Events(PAKE_LINK, (uintptr_t)info.m_V.usId, uRank, nullptr, CEventQueue::EVENT_TYPE_SYNC);
    if (uRet != RET_OKAY)
    {
        LOGV_WARN(m_FileLog, TF("[Slave]服务(RANK=%d)%p注册时业务实例检测id=%d返回错误=%d"), uRank, sSocket, info.m_V.usId, uRet);
        m_FileLog.Flush();
        return false;
    }

    PINDEX index = nullptr;
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PAIR_INFO* pPair = Cluster.Info.Find(sSocket);
        if (pPair == nullptr)
        {
            LOGV_INFO(m_FileLog, TF("[Slave]服务(TYPE=%d, RANK=%d)%p注册, Id=%d"), pLink->GetData(), uRank, sSocket, info.m_V.usId);
            index = Cluster.Info.Add(sSocket, info.m_V);
        }
        else if (pPair->m_V.usUpdate == (UShort)SERVERU_UNLINK)
        {
            // 注销的信息未同步到其他服务
            LOGV_INFO(m_FileLog, TF("[Slave]服务(TYPE=%d, RANK=%d)%p重新注册, Id=%d"), pLink->GetData(), uRank, sSocket, info.m_V.usId);
            pPair->m_V = info.m_V;
            index = reinterpret_cast<PINDEX>(pPair);
        }
        else
        {
            info.m_V.usUpdate = pPair->m_V.usUpdate;
        }
    }
    if (index != nullptr)
    {
        Cluster.Link(info.m_V);
        m_SrvData.IncrOnline();

        AddSync(uRank, info.m_V.uSync, sSocket);

        m_MgrHandler->OnHandle(PAKE_LINK, reinterpret_cast<uintptr_t>(&info), uRank);
        return true;
    }
    LOGV_WARN(m_FileLog, TF("[Slave]服务(RANK=%d)%p注册时存在异常状态=%d"), uRank, sSocket, info.m_V.usUpdate);
    m_FileLog.Flush();
    return false;
}

bool CSlaveServer::AddSync(UInt uRank, UInt uSync, Socket_t sSocket)
{
    if (uSync == 0)
    {
        return false;
    }
    for (UInt i = COMMONR_MIN; i < uRank; ++i)
    {
        if ((uSync & (1 << i)) != 0)
        {
            m_Sync[i].AddSync(sSocket);
        }
    }
    return true;
}

bool CSlaveServer::UpdateServer(UInt uRank, Socket_t sSocket, CNETHead* pUpdate)
{
    assert(pUpdate->IsTempStream());
    CStream& Stream = pUpdate->GetTempStream();

    PAIR_INFO info(sSocket);
    info.m_V.Status(Stream);
    assert(info.m_V.usUpdate == (UShort)SERVERU_SYNC);

    SERVER_STATUS ss;
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PAIR_INFO* pPair = Cluster.Info.Find(sSocket);
        if ( (pPair != nullptr) &&
             (pPair->m_V.usUpdate > (UShort)SERVERU_NONE) &&
             (pPair->m_V.usUpdate < (UShort)SERVERU_UNLINK) )
        {
            ss = pPair->m_V;
            pPair->m_V.Copy(info.m_V);
        }
        else if (pPair != nullptr)
        {
            ss.usLoad = pPair->m_V.usUpdate;
        }
    }
    if (ss.usUpdate > (UShort)SERVERU_NONE)
    {
        Cluster.Update(info.m_V, ss);
        m_MgrHandler->OnHandle(PAKE_UPDATE, reinterpret_cast<uintptr_t>(&info), uRank);
        return true;
    }
    if (ss.usLoad != (UShort)SERVERU_NONE)
    {
        LOGV_WARN(m_FileLog, TF("[Slave]服务(TYPE=%d, RANK=%d)%p更新时存在异常状态=%d"), pUpdate->GetData(), uRank, sSocket, ss.usLoad);
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Slave]服务(TYPE=%d, RANK=%d)%p更新时不存在"), pUpdate->GetData(), uRank, sSocket);
    }
    m_FileLog.Flush();
    return false;
}

bool CSlaveServer::SyncServer(UInt uRank, Socket_t sSocket, CPAKHead* pSync)
{
    LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)要求同步的层级编号%x"), pSync->GetRank(), uRank);
    bool bRet = true;
    SERVER_CLUSTER& Cluster = SyncCluster(uRank);
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PINDEX index = Cluster.Info.GetFirstIndex();
        do { bRet = SyncServerPack(Cluster, index, sSocket, uRank); } while (index != nullptr);
    }
    return bRet;
}

bool CSlaveServer::SyncServerPack(SERVER_CLUSTER& Cluster, PINDEX& index, Socket_t sSocket, UInt uRank)
{
    SYNCACK_DATA(SyncAck, uRank);
    if (SyncAck.Alloc(m_NetworkPtr, true))
    {
        CStream& Stream = SyncAck.GetStream(false);

        UInt uCount = 0;
        while (Cluster.CheckWrite(uCount, index, Stream, (UShort)SERVERU_OKAY_ALL));

        SyncAck.SetStatus(Cluster.Status);
        SERVER_STATUS& Status = SyncAck.GetStatus();
        Status.usRate   = (UShort)uCount; // 当前同步的数量
        Status.usUpdate = (index != nullptr) ? (UShort)SERVERU_NONE : (UShort)SERVERU_SYNC;

        LOGV_INFO(m_FileLog, TF("[Slave](RANK=%d)的集群状态向订阅的单个服务(socket=%p)同步%d服务信息"), uRank, sSocket, uCount);
        CNetStreamPtr StreamPtr;
        SyncAck.GetBuffer(m_NetworkPtr, StreamPtr);
        m_NetworkPtr->Send(sSocket, StreamPtr);
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Slave](RANK=%d)的集群状态向订阅的单个服务(socket=%p)同步服务信息申请缓存失败"), uRank, sSocket);
        m_FileLog.Flush();
    }
    return true;
}

bool CSlaveServer::UnlinkServer(UInt uRank, Socket_t sSocket, UInt uType)
{
    PAIR_INFO info(sSocket);
    SERVER_CLUSTER& Cluster = m_Cluster[uRank];
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        PAIR_INFO* pPair = Cluster.Info.Find(sSocket);
        if (pPair != nullptr)
        {
            info.m_V = pPair->m_V;
            if (info.m_V.usUpdate == (UShort)SERVERU_LINK)
            {
                // 服务从注册到注销, 信息还未同步到其他服务, 直接作为无状态信息删除
                Cluster.Info.RemoveAt(reinterpret_cast<PINDEX>(pPair));
            }
            else if (info.m_V.usUpdate != (UShort)SERVERU_UNLINK)
            {
                pPair->m_V.usUpdate = (UShort)SERVERU_UNLINK; // 需要同步给其他服务以后再删除
            }
        }
    }
    if (info.m_V.usUpdate != (UShort)SERVERU_NONE)
    {
        Cluster.Unlink(info.m_V);
        m_SrvData.DecrOnline();

        RemoveSync(uRank, info.m_V.uSync, sSocket);

        Events(PAKE_UNLINK, (uintptr_t)info.m_V.usId, uRank, nullptr, CEventQueue::EVENT_TYPE_SYNC);

        m_MgrHandler->OnHandle(PAKE_UNLINK, reinterpret_cast<uintptr_t>(&info), uRank);
        LOGV_INFO(m_FileLog, TF("[Slave]服务(TYPE=%d, RANK=%d)%p注销, Id=%d"), uType, uRank, sSocket, info.m_V.usId);
        return true;
    }
    LOGV_WARN(m_FileLog, TF("[Slave]服务(TYPE=%d, RANK=%d)%p注销时不存在"), uType, uRank, sSocket);
    m_FileLog.Flush();
    return false;
}

bool CSlaveServer::RemoveSync(UInt uRank, UInt uSync, Socket_t sSocket)
{
    if (uSync == 0)
    {
        return false;
    }
    for (UInt i = COMMONR_MIN; i < uRank; ++i)
    {
        if ((uSync & (1 << i)) != 0)
        {
            m_Sync[i].RemoveSync(sSocket);
        }
    }
    return true;
}
//--------------------------------------
bool CSlaveServer::OnTcpAccept(Socket_t sAccept, Socket_t)// sListen)
{
    if ((m_SrvData.nStatus == SERVERS_RUN) && (m_SrvData.IsOverLoad(false) == false))
    {
        CStrAddr strAddr;
        m_NetworkPtr->GetAddr(sAccept, strAddr.strIp, strAddr.usPort);
        LOGV_DUMP(m_FileLog, TF("[Slave]%s:%d连接完成!"), strAddr.strIp.GetBuffer(), strAddr.usPort);

        m_SrvData.IncrTotal();
        return true;
    }
    LOGV_WARN(m_FileLog, TF("[Slave]非运行状态STATUS=%d或者LOAD=%d!"), m_SrvData.nStatus, m_SrvData.usLoad);
    return false;
}
//--------------------------------------
bool CSlaveServer::OnTcpConnect(UInt uError, Socket_t sConnect)
{
    UInt uRank = (UInt)m_NetworkPtr->GetAttr(sConnect);
    assert(uRank < COMMONR_MAX);
    assert(m_SrvData.sConnect[uRank] == sConnect);
    assert(m_SrvData.nLink[uRank] == SERVERL_CNNTING);

    if (uError == 0)
    {
        m_SrvData.nLink[uRank] = SERVERL_CNNTED;

        Events(PAKE_TCPCONNECT, (uintptr_t)sConnect, uRank);
        m_MgrHandler->OnHandle(PAKE_TCPCONNECT, (uintptr_t)sConnect, uRank);

        LinkServer(uRank);
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Slave]连接内部服务(RANK=%d)失败%X!"), uRank, uError);
    }
    return true;
}
//--------------------------------------
bool CSlaveServer::OnTcpClose(Socket_t sSocket, ULLong ullLiveData)
{
    if (CAtomics::CompareExchange<UInt>((PUInt)&m_SrvData.nStatus, 0, 0) == SERVERS_STOP)
    {
        return false;
    }
    if (sSocket == m_SrvData.sListen)
    {
        m_SrvData.sListen = 0;
        LOG_WARN(m_FileLog, TF("[Slave]监听关闭, 进入告警状态"));
        m_FileLog.Flush();

        if (m_SrvData.nStatus < SERVERS_ERROR)
        {
            m_SrvData.SetWarnLoad();
        }
        Events(PAKE_TCPLISTEN_CLOSE, (uintptr_t)sSocket, m_SrvData.uRank);
        m_MgrHandler->OnHandle(PAKE_TCPLISTEN_CLOSE, (uintptr_t)sSocket, m_SrvData.uRank);
    }
    else if (ullLiveData != 0)
    {
        if (ullLiveData < COMMONR_MAX) // Connect-Socket
        {
            UInt uRank = (UInt)ullLiveData;
            assert(m_SrvData.sConnect[uRank] == sSocket);
            assert(m_SrvData.nLink[uRank] >= SERVERL_READY);

            Int nLink = m_SrvData.nLink[uRank];

            m_SrvData.sConnect[uRank] = 0;
            m_SrvData.nLink[uRank] = SERVERL_READY;

            if ((uRank == COMMONR_MASTER) && (nLink >= SERVERL_LINKED))
            {
                CleanCluster();
            }

            if (nLink >= SERVERL_CNNTED)
            {
                Events(PAKE_TCPCONNECT_CLOSE, (uintptr_t)sSocket, uRank);
                m_MgrHandler->OnHandle(PAKE_TCPCONNECT_CLOSE, (uintptr_t)sSocket, uRank);
            }
        }
        else
        {
            // 单个服务断开
            ullLiveData &= 0x0000FFFFFFFFFFFFULL; // remove high slave-rank-16
            UnlinkServer((UInt)ullLiveData, sSocket, (UInt)(ullLiveData >> 32));
            Events(PAKE_TCPACCEPT_CLOSE, (uintptr_t)sSocket, ullLiveData);
            m_SrvData.DecrTotal();
        }
    }
    else
    {
        m_SrvData.DecrTotal();
    }
    return true;
}
//--------------------------------------
void CSlaveServer::LinkServer(UInt uRank)
{
    Int& nLink = m_SrvData.nLink[uRank];
    assert(nLink == SERVERL_CNNTED);

    NET_DATA(Link, PAKE_LINK, COMMONT_SLAVE);
    if (Link.Alloc(m_NetworkPtr, true))
    {
        CStream& Stream = Link.GetStream(false);
        m_SrvData.Status(Stream, (UShort)SERVERU_OKAY_ALL);
        if (uRank == COMMONR_MASTER)
        {
            m_SrvData.Info(Stream);
        }
        else
        {
            m_SrvData.Info(Stream, (UShort)(SERVERU_SYNC_ALL|SERVERU_SYNCMASK));
        }

        nLink = SERVERL_LINKING;
        LOGV_INFO(m_FileLog, TF("[Slave]向连接的内部服务(RANK=%d)注册服务"), uRank);
        CNetStreamPtr StreamPtr;
        Link.GetBuffer(m_NetworkPtr, StreamPtr);
        m_NetworkPtr->Send(m_SrvData.sConnect[uRank], StreamPtr);
    }
    else
    {
        LOGV_WARN(m_FileLog, TF("[Slave]向连接的内部服务(RANK=%d)注册服务时申请缓存失败"), uRank);
        m_FileLog.Flush();
    }
}

void CSlaveServer::UnlinkServer(UInt uRank)
{
    if (m_SrvData.nLink[uRank] == SERVERL_LINKED)
    {
        LOGV_INFO(m_FileLog, TF("[Slave]向连接的内部服务(RANK=%d)注销服务"), uRank);
        PAK_DATA(Unlink, PAKE_UNLINK, COMMONT_SLAVE);
        m_NetworkPtr->Send(m_SrvData.sConnect[uRank], Unlink);
    }
}
//--------------------------------------
void CSlaveServer::OnLinkAck(UInt uRank, CPAKHead* pLinkAck, Socket_t sSocket)
{
    LOGV_INFO(m_FileLog, TF("[Slave]收到内部服务(RANK=%d)注册回复包"), uRank);
    m_SrvData.nLink[uRank] = SERVERL_LINKED;
    if (uRank == COMMONR_MASTER)
    {
        // 注册返回数据是在线数量, 保存主同步服务的数值
        m_Cluster[m_SrvData.uRank].Status.uOnline = pLinkAck->GetData();
        // 拉取需要同步的服务数据
        if (m_SrvData.uSync != 0)
        {
            LOGV_INFO(m_FileLog, TF("[Slave]收到同步服务(RANK=%d)注册回复包, 请求需要同步的层级比特掩码%x"), uRank, m_SrvData.uSync);
            PAK_DATA(Sync, PAKE_SYNC, m_SrvData.uSync);
            m_NetworkPtr->Send(sSocket, Sync);
        }
        if (m_SrvData.nStatus == SERVERS_RUN)
        {
            // 注册完成, 切换到正常状态
            m_SrvData.Clear();
            if (m_SrvData.IsWarnLoad(false) == false)
            {
                m_SrvData.usLoad += m_pConfig->GetServerLoad(m_SrvData.uRank);
            }
        }
    }
    m_MgrHandler->OnHandle(PAKE_LINKACK, (uintptr_t)sSocket, uRank);
}

void CSlaveServer::OnSyncAck(UInt uRank, CNETSync* pSyncAck, Socket_t sSocket)
{
    LOGV_INFO(m_FileLog, TF("[Slave]收到内部服务(RANK=%d)同步的(SYNC-RANK=%d)的服务信息Count=%d[%x]"),
              uRank, pSyncAck->GetData(), pSyncAck->GetStatus().usRate, pSyncAck->GetStatus().usUpdate);
    uRank = pSyncAck->GetData(); // 同步的服务层级
    SERVER_CLUSTER& Cluster = m_Master[uRank];

    Cluster.Status = pSyncAck->GetStatus();
    if (Cluster.Status.usRate > 0) // 当前同步的数量
    {
        CSyncLockWaitScope scope(Cluster.Lock);
        assert(pSyncAck->IsTempStream());
        CStream& Stream = pSyncAck->GetTempStream();
        while (Stream.IsEnd() == false)
        {
            Cluster.Read(Stream);
        }
    }
    // 同步完成
    if (Cluster.Status.usUpdate == (UShort)SERVERU_SYNC)
    {
        m_MgrHandler->OnHandle(PAKE_SYNCACK, reinterpret_cast<uintptr_t>(&Cluster), uRank);
        //// 同步给其他服务再更新状态
        // {
        //     CSyncLockWaitScope scope(Cluster.Lock);
        //     Cluster.Update();
        //     Cluster.Status.usUpdate == (UShort)SERVERU_OKAY;
        // }
    }
}

void CSlaveServer::OnUnlinkAck(UInt uRank, Socket_t sSocket)
{
    m_SrvData.nLink[uRank] = SERVERL_CNNTED;
    if (uRank == COMMONR_MASTER)
    {
        CleanCluster();
    }
    m_MgrHandler->OnHandle(PAKE_UNLINKACK, (uintptr_t)sSocket, uRank);
    LOGV_INFO(m_FileLog, TF("[Slave]收到内部服务(RANK=%d)注销回复包"), uRank);
}
